package com.atguigu.edu.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.util.DimUtil;
import com.atguigu.edu.realtime.util.DruidPhoenixDSUtil;
import com.atguigu.edu.realtime.util.JedisUtil;
import com.atguigu.edu.realtime.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @author Lec
 * @date 2022/8/17 13:27
 */

public class MyPhoenixSink extends RichSinkFunction<JSONObject> {
    private DruidDataSource druidDataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidPhoenixDSUtil.getDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) {
        // 如果维度表的数据类型为update删除掉redis中对应的缓存数据  保持数据一致性
//        String type = value.getString("type");
////        Jedis jedis = JedisUtil.getJedis();
////        String redisKey = "DIM:" + value.getString("sink_table").toUpperCase() + ":" + value.getString("sku_id");
//
//        if ("update".equals(type)) {
//            String tableName = value.getString("sink_table").toUpperCase();
//            String id = value.getString("id");    /*这里是id不是sku_id*/
//            DimUtil.deleteRedisCache(tableName, id);
//        }
////        jedis.close();
//
//        /*----------------防止下面一步将type给当做列名写入-----------------*/
        value.remove("type");


        String sql = createSql(value);

        System.out.println(sql);

        DruidPooledConnection connection = null;
        try {
            connection = druidDataSource.getConnection();
            PhoenixUtil.executeSql(sql, connection);
//            System.out.println("111111111111111");
        } catch (SQLException e) {
            e.printStackTrace();
            System.out.println("获取druid链接异常");
        }

    }

    private String createSql(JSONObject value) {
        String sink_table = value.getString("sink_table");

        value.remove("sink_table");
        Set<String> keySet = value.keySet();
        Collection<Object> values = value.values();

        StringBuilder sql = new StringBuilder();
//        UPSERT INTO TEST(NAME,ID) VALUES('foo',123);
        sql.append("upsert into ")
                .append(EduConfig.HBASE_SCHEMA)
                .append(".")
                .append(sink_table)
                .append(" ( ")
                .append(StringUtils.join(keySet, ","))
                .append(" ) ")
                .append("values")
                .append(" ( '")
                .append(StringUtils.join(values, "' , '"))
                .append("') ");

        return sql.toString();

    }
}
